package com.mq.mqserver.datacore;

import com.mq.common.BinaryTool;
import com.mq.common.MqException;
import com.mq.mqserver.core.MSGQueue;
import com.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/**
 * 消息文件管理类:针对硬盘上的文件的操作
 * 文件不能只存储在内存中,要将其存储在磁盘上,这样才能做到持久化存储
 * 每一个文件存放两个子文件:消息数据文件+消息统计文件(消息数据文件中总的消息数\t有效的消息数)
 */
public class MessageFileManager {
    //使用静态内部类表示 :消息的 统计信息
    public static class Stat{
        public int totalCount;//文件中总的消息数量
        public int validCount;//文件中有效消息的数量
    }
    //初始化该类:目前暂无逻辑要执行
    public void init(){}
    //这个方法用来获取指定 队列对应的消息文件所在的路径
    public String getQueueDir(String queueName){
        return "./data/"+queueName;
    }

    //获取指定队列对应的 消息数据文件 所在的路径
    //存储消息内容的文件是以二进制的形式存储的,一般都是以.bin .dat,这里就用.txt为后缀了,暂时不变了
    private String getDataFilePath(String queueName){
        return getQueueDir(queueName)+"/queue_data.txt";
    }
    //获取指定队列对应的 消息统计文件所在的路径
    //这里存储消息统计信息的文件很短,用文本文件存储
    private String getStatFilePath(String queueName){
        return getQueueDir(queueName)+"/queue_stat.txt";
    }

    //读取消息统计文件中的内容:这个文件是文本文件,直接用Scanner来读取即可
    private Stat readStat(String queueName){
        Stat stat = new Stat();
        try (InputStream inputStream = new FileInputStream(getStatFilePath(queueName))){
            //使用Scanner将FileInputStream类封装起来,进行读取
            Scanner scan = new Scanner(inputStream);
            stat.totalCount = scan.nextInt();
            stat.validCount = scan.nextInt();
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    //往统计文件中写信息
    //使用PrintWrite来写数据到文件
    private void writeStat(String queueName,Stat stat){
        //FileOutputStream默认会在打开文件的时候,将文件内容清空,
        // 可以通过设置第二个参数为true,就会以追加的形式写入数据到文件,不会清空原来文件的内容
        //这里还是使用第一种形式打开文件为空文件的方式
        try(OutputStream outputStream = new FileOutputStream(getStatFilePath(queueName))){
            PrintWriter printWrite = new PrintWriter(outputStream);
            printWrite.write(stat.totalCount+"\t"+stat.validCount);
            //要刷新文件,
            //IO草最都是很耗时的,因此会将要写入文件的内容先放到一个缓冲区中,
            // 直到该缓冲区写满或满足其他条件才将数据一次性写入到文件中
            // 为防止数据还存在未满的缓冲区中,
            // 因此 当一次数据写入完成后,通过刷新缓冲区将文件写入到文件中,确保数据都存入到了文件中
            printWrite.flush();
        } catch (IOException e) {
            e.printStackTrace();
//            throw new RuntimeException(e);
        }
    }
    //创建队列 对应的文件和目录
        //1.创建对应的目录
        //2.创建对应的文件
        //3.初始化统计文件
    public void createQueueFile(String queueName) throws IOException {
        //1.创建对应的目录
        File baseDir = new File(getQueueDir(queueName));
        if(!baseDir.exists()){
            //当目录不存在时,就用该目录名创建
            boolean ok = baseDir.mkdirs();
            if(!ok){
                throw new IOException("[createQueueFile] 创建目录失败!"+ baseDir.getAbsolutePath());
            }
        }
        //2.创建对应的文件:数据文件,统计文件
        File dataFile = new File(getDataFilePath(queueName));
        if(!dataFile.exists()){
            //数据文件不存在,就创建文件,
            //这里是要创建文件,而不是创建目录******
            boolean ok = dataFile.createNewFile();
            if(!ok){
                throw new IOException("[createQueueFile] 创建数据文件失败!"+dataFile.getAbsolutePath());
            }
        }
        //2.创建对应的文件:统计文件
        File statFile = new File(getStatFilePath(queueName));
        if(!statFile.exists()){
            //统计文件不存在,就创建文件
            boolean ok = statFile.createNewFile();
            if(!ok){
                throw new IOException("[createQueueFile] 创建统计文件失败!"+statFile.getAbsolutePath());
            }
        }
        //初始化统计文件,设置初值
        Stat stat = new Stat();
        stat.totalCount = 0;
        stat.validCount = 0;
        writeStat(queueName , stat);
    }

    //删除指定队列的目录和文件
    //当队列被删除后,对应的文件和目录也就要被删除了
    public void destoryQueueFile(String queueName) throws IOException {
        //1.删除数据文件
        File dataFile = new File(getDataFilePath(queueName));
        boolean ok1 = dataFile.delete();
        //2.删除统计文件
        File statFile = new File(getStatFilePath(queueName));
        boolean ok2 = statFile.delete();
        //3.删除总目录
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if(!ok1 || !ok2 || !ok3){
            throw new IOException("删除队列和文件失败! baseDir: "+baseDir.getAbsolutePath());
        }
    }

    //检查目录和文件是否存在
    public boolean checkFileExists(String queueName){
        File baseDir = new File(getQueueDir(queueName));
        if(!baseDir.exists()){
            return false;
        }
        File dataFile = new File(getDataFilePath(queueName));
        if(!dataFile.exists()){
            return false;
        }
        File statFile = new File(getStatFilePath(queueName));
        if(!statFile.exists()){
            return false;
        }
        return true;
    }
//    //将新的消息写入到队列对应的文件中
//    public void sendMessage1(MSGQueue queue,Message message) throws MqException, IOException {
//        boolean ok = checkFileExists(queue.getName());
//        if(!ok){
//            throw new MqException("文件不存在!");
//        }
//        //将消息序列化
//        byte[] byteData = BinaryTool.toByte(message);
//        int len = byteData.length;
//        synchronized(queue){
//            File file = new File(getDataFilePath(queue.getName()));
//            message.setOffsetBeg(file.length()+4);
//            message.setOffsetEnd(file.length()+4+len);
//            //写入到文件中,以追加的形式
//            FileOutputStream outputStream = new FileOutputStream(getDataFilePath(queue.getName()),true);
//            DataOutputStream dataOutputStream = new DataOutputStream(outputStream);
//            dataOutputStream.writeInt(len);
//            dataOutputStream.write(byteData);
//            //修改统计文件数据
//            Stat stat = readStat(queue.getName());
//            stat.totalCount+=1;
//            stat.validCount+=1;
//            writeStat(queue.getName(),stat);
//        }
//    }
//
//    //从文件中奖消息删除,逻辑的删除
//    public void deleteMessageFromFile1(MSGQueue queue,Message message) throws IOException, ClassNotFoundException {
//        synchronized(queue){
//            //要是随机读取才行
//            RandomAccessFile randomAccessFile = new RandomAccessFile(getDataFilePath(queue.getName()),"rw");
//            //获取读取文件的长度
//            long len = message.getOffsetEnd()-message.getOffsetBeg();
//            byte[] dataByte = new byte[(int)len];
//            randomAccessFile.seek(message.getOffsetBeg());
//            randomAccessFile.read(dataByte);
//            //反序列化为对象
//            Message newMessage = (Message) BinaryTool.fromBytes(dataByte);
//            //修改为无效
//            newMessage.setIsValid((byte)0x0);
//            //再将其写入到文件中
//            //先将对象序列化为字节数字
//            byte[] newByteData = BinaryTool.toByte(newMessage);
//            randomAccessFile.seek(message.getOffsetBeg());
//            randomAccessFile.write(newByteData);
//            //更新统计文件中的数据
//            Stat stat = readStat(queue.getName());
//            if(stat.validCount>0){
//                stat.validCount-=1;
//            }
//            writeStat(queue.getName(),stat);
//        }
//    }
//    //加载文件中的所有消息到内存中(链表)
//    public LinkedList<Message> loadAllMessageFromFile(String queueName) throws IOException, MqException, ClassNotFoundException {
//        LinkedList<Message> list = new LinkedList<>();
//        //先读取文件
//        try(FileInputStream fileInputStream = new FileInputStream(queueName)){
//            try(DataInputStream dataInputStream = new DataInputStream(fileInputStream)){
//                int currentOffset = 0;
//                while(true){
//                    int len = dataInputStream.readInt();
//                    byte[] dataByte = new byte[len];
//                    int actualLen = dataInputStream.read(dataByte);
//                    if(actualLen!=len){
//                        throw new MqException("文件中消息格式错误!");
//                    }
//                    //将字节数组反序列化为对象
//                    Message newMessage = (Message) BinaryTool.fromBytes(dataByte);
//                    if(newMessage.getIsValid()==0x0){
//                        //消息为无效的
//                        currentOffset+=(4+len);
//                        continue;
//                    }
//                    newMessage.setOffsetBeg(currentOffset+4);
//                    newMessage.setOffsetEnd(currentOffset+4+len);
//                    currentOffset+=(4+len);
//                    list.add(newMessage);
//                }
//            }catch (EOFException e){
//                System.out.println("文件读取到末尾!");
//            }
//        }
//        return list;
//    }

    //将一条新的消息写入到 队列对应的 文件中
    //就是将消息保存到硬盘上
    //需要先将消息对象序列化成二进制字节数组,然后写入到对应的文件中
    //会涉及到多线程安全问题:在设置消息在文件中的位置和修改统计文件中的数据的时候,都有可能出现线程安全问题
    //这主要是多个线程对同一个队列操作造成的,因此要对队列上锁
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        boolean exists = checkFileExists(queue.getName());
        if(!exists){
            throw new MqException("[MessageFileManager] 队列对应的文件不存在, queueName: "+queue.getName());
        }
        //1.将消息对象序列化成二进制字节数组
        byte[] byteMessage = BinaryTool.toByte(message);
        synchronized(queue){//对队列进行上锁,当多个线程访问同一个队列的时候,只能有一个线程获取到操作队列的权限
            //2. 将其写入到数据文件中
            // 获取数据文件路径
            File dataFile = new File(getDataFilePath(queue.getName()));
            //设置消息在文件中的始终位置
            message.setOffsetBeg(dataFile.length()+4);
            message.setOffsetEnd(dataFile.length()+4+byteMessage.length);
            //将消息写入到文件中
            //  是以追加的形式写入,将第二个参数设为true
            try(FileOutputStream outputStream = new FileOutputStream(dataFile,true)){
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    //先写前4个字节:记录消息的长度,
                    //注意,这里规定的就是用4字节的长度,不能单使用write写入,这样还是只写入1个字节
                    dataOutputStream.writeInt(byteMessage.length);
                    //写消息体到文件
                    dataOutputStream.write(byteMessage);
                }
            }
            //3.更新统计文件数据
            Stat stat = readStat(queue.getName());
            stat.totalCount+=1;
            stat.validCount+=1;
            writeStat(queue.getName(),stat);
        }
    }
    //将一条消息从队列对应的文件中删除
    //这里采用的是物理删除,仅将文件中消息的isValid置为0x0即可,而不是真的删除
    //现将消息从文件中读取出来,然后修改isValid属性,然后再将其写入到文件中
    //删除操作也存在线程安全问题:要给队列上锁
    public void deleteMessageFromFile(MSGQueue queue,Message message) throws IOException, ClassNotFoundException {
        synchronized(queue) {
            //1.将消息从文件中取出来
            //需要找到消息在文件中的位置,然后读取,要随机读取,因此要用RandomAccessFile类,且以可读可写的方式打开
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(getDataFilePath(queue.getName()), "rw")) {
                //先准备一个装消息的数组容器:
                byte[] data = new byte[(int) (message.getOffsetEnd() - message.getOffsetBeg())];
                //将光标定位到 文件中数据所在的起始位置
                randomAccessFile.seek(message.getOffsetBeg());
                //读取数据到准备好的字节数组容器中
                randomAccessFile.read(data);
                //2.将读取的二进制数据转换成对象
                Message newMessage = (Message) BinaryTool.fromBytes(data);
                //3.设置isValid为无效
                newMessage.setIsValid((byte) 0x0);
                //4.重新写入文件
                //将消息对象先转换为字节数组
                byte[] newByte = BinaryTool.toByte(newMessage);
                //重新写入文件
                //重新获取光标,在读文件或写文件都会修改光标的位置,因此要重新获取光标的插入位置
                randomAccessFile.seek(message.getOffsetBeg());
                randomAccessFile.write(newByte);
                //5.修改统计文件数据
                Stat stat = readStat(queue.getName());
                if (stat.validCount > 0) {
                    stat.validCount -= 1;
                }
                writeStat(queue.getName(), stat);
            }
        }
    }

    //读取文件中所有的消息到内存中(将数据从磁盘加载到内存(链表形式存储)中)
    //用于当MQ服务器宕机重启后的消息回复
    //这个方法不存在线程安全问题,因此在传参数的时候,由于不涉及到对队列的加锁,只传队列名即可
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
       //使用链表存储读取到的消息集合
        LinkedList<Message> messages = new LinkedList<>();
        try(FileInputStream fileInputStream = new FileInputStream(getDataFilePath(queueName))){
            try(DataInputStream dataInputStream = new DataInputStream(fileInputStream)){
                //记录读取文件的起始位置
                long currentOfferSet = 0;
                //消息很多,需要一直读取
               while(true){
                   //1.从文件中先读取一条消息的长度
                    int len = dataInputStream.readInt();
                    //2.准备装数据的容器
                    byte[] byteData = new byte[len];
                    //3.读取文件
                    int actualLen = dataInputStream.read(byteData);
                    if(actualLen!=len){
                        //4.当实际读取到的数据长度和预期的数据长度不一致时,说明该数据存在问题,格式乱了
                        throw new MqException("[MessageFileManager] 文件中的格式错误 queueName: "+queueName);
                    }
                    //5.将读取到的数据转化成对象
                   Message message = (Message) BinaryTool.fromBytes(byteData);
                   //6.判断是否为有效数据,若数据为无效,直接跳过
                   byte ok = message.getIsValid();
                   if(ok==0x0){
                       currentOfferSet+=(4+len);
                       continue;
                   }
                    //7.记录消息在文件中的位置信息
                    message.setOffsetBeg(currentOfferSet+4);
                   message.setOffsetEnd(currentOfferSet+4+len);
                   currentOfferSet+=(4+len);
                   //存到集合中
                   messages.add(message);
               }
            }catch(EOFException e){
                //执行到这里是文件读取结束了,读取到文件末尾,会被readInt抛出异常
                //属于整场的执行逻辑
                System.out.println("[MessageFileManager] 恢复message完成!");
            }
        }
        return messages;
    }

    //进行垃圾消息的回收:
    // 创建一个新的文件,将要进行GC的文件中的有效消息都搬到新文件中,再把就文件删除,
    // 将新文件名改为删除的文件名,并修改统计文件数据,GC完成.
    //检查当前文件是否需要进行GC:文件中有效消息占总消息比例<50% && 总消息数大于2000
    public boolean checkGC(String queueName){
        Stat stat = readStat(queueName);
        int totalCount = stat.totalCount;
        int validCount = stat.validCount;
        if(totalCount>=2000 && validCount * 1.0 / totalCount <0.5){
            return true;
        }
        return false;
    }
    //创建一个新的文件名
    private String getNewDataFilePath(String queueName) throws IOException {
        return getQueueDir(queueName)+"queue_data_new.txt";
    }

    //进行垃圾回收
    public void GC(MSGQueue queue) throws IOException, MqException, ClassNotFoundException {
        synchronized(queue){
        //垃圾回收是需要对消息进行大洗牌,比较耗时,记录一下消耗时间,以防出现问题,及时定位出错原因
            long GCBegTime = System.currentTimeMillis();
            //1.创建一个新文件
            File newDataFile = new File(getNewDataFilePath(queue.getName()));
            //正常情况下,新文件是不存在的,若已经存在,说明上次GC未执行成功,留下了残留文件
            if(newDataFile.exists()){
                throw new MqException("[MessageFileManager] 上次GC回收未执行成功,存在残留文件 queueName:"+queue.getName());
            }
            boolean ok = newDataFile.createNewFile();
            if(!ok){
                throw new MqException("[MessageFileManager] 创建新文件失败!fileName: "+ newDataFile.getAbsolutePath());
            }
            //将旧文件中的有效消息都取出来
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());
            //将对象进行序列化,再存入新的文件中
            try(FileOutputStream outputStream = new FileOutputStream(newDataFile)){
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)){
                    //依次读取每一条消息
                    for(Message message:messages){
                        byte[] byteMessage = BinaryTool.toByte(message);
                        //存入新文件中
                        int len = byteMessage.length;
                        dataOutputStream.writeInt(len);
                        dataOutputStream.write(byteMessage);
                    }
                }
            }
            //删除旧文件
            File oldQueueFile = new File(getDataFilePath(queue.getName()));
            ok = oldQueueFile.delete();
            if(!ok){
                throw new MqException("[MessageFileManager] 删除旧文件失败 oldFilePath: "+oldQueueFile.getAbsolutePath());
            }
            //将新文件的文件名修改为旧文件的文件名
            ok = newDataFile.renameTo(oldQueueFile);
            if(!ok){
                throw new MqException("[MessageFileManager] 新文件改名失败 newDataFile: "+newDataFile.getAbsolutePath());
            }
            //修改统计文件数据
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(),stat);

            long GCEndTime = System.currentTimeMillis();
            long def = GCEndTime - GCBegTime;
            System.out.println("本次GC消耗的时间为: "+def+"ms");
        }
    }

}
